package com.example.literj.kt.internal.operators.single

import com.example.literj.kt.core.Scheduler
import com.example.literj.kt.core.Single
import com.example.literj.kt.core.SingleObserver
import com.example.literj.kt.core.SingleSource
import com.example.literj.kt.disposables.Disposable
import com.example.literj.kt.internal.disposables.DisposableHelper
import com.example.literj.kt.internal.disposables.SequentialDisposable
import java.util.concurrent.atomic.AtomicReference

class SingleSubscribeOn<T>(
    private val source: SingleSource<out T>,
    private val scheduler: Scheduler
) : Single<T>() {
    override fun subscribeActual(observer: SingleObserver<in T>) {
        val parent = SubscribeOnObserver(observer, source)
        observer.onSubscribe(parent)
        val f = scheduler.scheduleDirect(parent)
        parent.task.replace(f)
    }

    internal class SubscribeOnObserver<T>(
        private val downstream: SingleObserver<in T>,
        private val source: SingleSource<out T>
    ) :
        AtomicReference<Disposable>(), SingleObserver<T>, Disposable,
        Runnable {
        val task = SequentialDisposable()

        override fun onSubscribe(d: Disposable) {
            DisposableHelper.setOnce(this, d)
        }

        override fun onSuccess(t: T) {
            downstream.onSuccess(t)
        }

        override fun onError(e: Throwable) {
            downstream.onError(e)
        }

        override fun dispose() {
            DisposableHelper.dispose(this)
            task.dispose()
        }

        override fun isDisposed() = DisposableHelper.isDisposed(get())

        override fun run() {
            source.subscribe(this)
        }
    }
}